package com.study.chapter08;

import com.study.entity.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @Description:
 * @Author: LiuQun
 * @Date: 2022/8/12 20:49
 */
public class IntervalJoinTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Tuple2<String,Long>> stream1 = env.fromElements(
                Tuple2.of("Bob",1000L),
                Tuple2.of("Mary",2000L),
                Tuple2.of("Alice",2000L),
                Tuple2.of("Livy",3000L)

        ).assignTimestampsAndWatermarks(
                WatermarkStrategy.<Tuple2<String,Long>>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner((data, l) -> data.f1)
        );

        SingleOutputStreamOperator<Event> stream2 = env.fromElements(
                new Event("Bob", "./home", 1000L),
                new Event("Livy", "./home", 2000L),
                new Event("Alice", "./cart", 3000L),
                new Event("Bob", "./cart", 2000L),
                new Event("Livy", "./prod?id=20", 4000L),
                new Event("Bob", "./prod?id=30", 5000L),
                new Event("Bob", "./home", 6000L),
                new Event("Alice", "./home/car", 7000L)
        ).assignTimestampsAndWatermarks(
                WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner((data, l) -> data.timestamp)
        );

       stream1.keyBy(tup -> tup.f0)
               .intervalJoin(stream2.keyBy(event -> event.user))
               .between(Time.seconds(-5),Time.seconds(10))
               .process(new ProcessJoinFunction<Tuple2<String, Long>, Event, String>() {
                   @Override
                   public void processElement(Tuple2<String, Long> left, Event right, Context ctx, Collector<String> out) throws Exception {
                       out.collect(right + "=>" + left);
                   }
               })
               .print();


        env.execute();
    }
}
